{
 "cells": [
  {
   "cell_type": "code",
   "id": "initial_id",
   "metadata": {
    "collapsed": true
   },
   "source": [
    ""
   ],
   "outputs": [],
   "execution_count": null
  },
  {
   "metadata": {
    "ExecuteTime": {
     "end_time": "2025-06-09T12:14:00.580135Z",
     "start_time": "2025-06-09T12:13:10.526028Z"
    }
   },
   "cell_type": "code",
   "source": [
    "import random\n",
    "import json\n",
    "from datetime import datetime\n",
    "from time import sleep\n",
    "from kafka import KafkaProducer\n",
    "from kafka.errors import KafkaError\n",
    "\n",
    "# Kafka配置\n",
    "KAFKA_BOOTSTRAP_SERVERS = 'node101:9092,node102:9092,node103:9092'\n",
    "TOPIC_NAME = 'user_profile_tags'\n",
    "\n",
    "# 初始化Kafka生产者\n",
    "producer = KafkaProducer(\n",
    "    bootstrap_servers=KAFKA_BOOTSTRAP_SERVERS,\n",
    "    value_serializer=lambda v: json.dumps(v).encode('utf-8'),\n",
    "    acks='all',\n",
    "    retries=3\n",
    ")\n",
    "\n",
    "# 数据生成配置\n",
    "age_groups = ['18-24岁', '25-29岁', '30-34岁', '35-39岁', '40-49岁', '50岁以上']\n",
    "gender_weights = {'女性用户': 45, '男性用户': 45, '家庭用户': 10}\n",
    "\n",
    "def generate_user_record(seq):\n",
    "    \"\"\"生成单条用户记录（带数据分布逻辑）\"\"\"\n",
    "    user_id = f\"u{seq:08d}\"\n",
    "    \n",
    "    # 修正的年龄标签生成（带群体分布）\n",
    "    age_idx = min(int(abs(random.gauss(2, 1.5))), 5)  # 修正的括号\n",
    "    age_group = age_groups[age_idx]\n",
    "    \n",
    "    # 计算出生年份\n",
    "    age_range = age_group.split('岁')[0].split('-')\n",
    "    min_age = int(age_range[0])\n",
    "    max_age = int(age_range[1]) if len(age_range) > 1 else 70\n",
    "    birth_year = 2023 - random.randint(min_age, max_age)\n",
    "    \n",
    "    # 性别标签（带权重）\n",
    "    gender = random.choices(\n",
    "        list(gender_weights.keys()),\n",
    "        weights=gender_weights.values()\n",
    "    )[0]\n",
    "    \n",
    "    # 身体指标（带相关性）\n",
    "    base_height = 158 if '女性' in gender else 170\n",
    "    height = max(140, min(200, int(random.gauss(base_height, 5))))\n",
    "    weight = max(35, min(100, int(random.gauss(height/2.5, 3))))\n",
    "    \n",
    "    # 星座生成\n",
    "    def get_constellation(month, day):\n",
    "        dates = [\n",
    "            (3, 21, 4, 19), (4, 20, 5, 20), (5, 21, 6, 21),\n",
    "            (6, 22, 7, 22), (7, 23, 8, 22), (8, 23, 9, 22),\n",
    "            (9, 23, 10, 23), (10, 24, 11, 22), (11, 23, 12, 21),\n",
    "            (12, 22, 1, 19), (1, 20, 2, 18), (2, 19, 3, 20)\n",
    "        ]\n",
    "        constellations = [\n",
    "            '白羊座', '金牛座', '双子座', '巨蟹座', '狮子座', \n",
    "            '处女座', '天秤座', '天蝎座', '射手座', '摩羯座', '水瓶座', '双鱼座'\n",
    "        ]\n",
    "        for i, (start_m, start_d, end_m, end_d) in enumerate(dates):\n",
    "            if (month == start_m and day >= start_d) or (month == end_m and day <= end_d):\n",
    "                return constellations[i]\n",
    "        return constellations[0]\n",
    "    \n",
    "    month = random.randint(1, 12)\n",
    "    day = random.randint(1, 28)\n",
    "    constellation = get_constellation(month, day)\n",
    "    \n",
    "    # 10%概率生成设备实时数据\n",
    "    is_device_data = random.random() < 0.1\n",
    "    update_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')\n",
    "    \n",
    "    return {\n",
    "        \"user_id\": user_id,\n",
    "        \"age_label\": age_group,\n",
    "        \"birth_year\": birth_year,\n",
    "        \"gender_label\": gender,\n",
    "        \"weight_kg\": round(weight + random.random(), 1),\n",
    "        \"height_cm\": height,\n",
    "        \"constellation\": constellation,\n",
    "        \"data_source\": \"device\" if is_device_data else \"system\",\n",
    "        \"update_time\": update_time,\n",
    "        \"metadata\": {\n",
    "            \"generation_seq\": seq,\n",
    "            \"batch_id\": datetime.now().strftime('%Y%m%d%H%M')\n",
    "        }\n",
    "    }\n",
    "\n",
    "def send_to_kafka(record):\n",
    "    \"\"\"异步发送消息到Kafka（带回调处理）\"\"\"\n",
    "    try:\n",
    "        future = producer.send(\n",
    "            TOPIC_NAME,\n",
    "            key=record['user_id'].encode('utf-8'),\n",
    "            value=record\n",
    "        )\n",
    "        \n",
    "        def callback(err, msg):\n",
    "            if err:\n",
    "                print(f\"消息发送失败: {err}\")\n",
    "            else:\n",
    "                if int(record['metadata']['generation_seq']) % 1000 == 0:\n",
    "                    print(f\"成功发送: {msg.topic}[{msg.partition}]@{msg.offset}\")\n",
    "        \n",
    "        future.add_callback(callback).add_errback(\n",
    "            lambda err: print(f\"发送错误: {err}\")\n",
    "        )\n",
    "    except Exception as e:\n",
    "        print(f\"发送异常: {e}\")\n",
    "\n",
    "def generate_and_send_batch(batch_size=1000):\n",
    "    \"\"\"批量生成并发送数据\"\"\"\n",
    "    for i in range(1, batch_size + 1):\n",
    "        record = generate_user_record(i)\n",
    "        send_to_kafka(record)\n",
    "        \n",
    "        # 控制发送速率（约1000条/秒）\n",
    "        if i % 1000 == 0:\n",
    "            producer.flush()\n",
    "            print(f\"已生成 {i} 条记录\")\n",
    "            sleep(0.5)\n",
    "    \n",
    "    # 确保所有消息都已发送\n",
    "    producer.flush()\n",
    "    print(f\"总共发送 {batch_size} 条记录\")\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "    try:\n",
    "        print(f\"开始向Kafka主题 {TOPIC_NAME} 发送数据...\")\n",
    "        generate_and_send_batch(50000)\n",
    "        print(\"数据发送完成！\")\n",
    "    except KafkaError as e:\n",
    "        print(f\"Kafka错误: {e}\")\n",
    "    except KeyboardInterrupt:\n",
    "        print(\"用户中断操作\")\n",
    "    finally:\n",
    "        producer.close()\n",
    "        print(\"生产者已关闭\")"
   ],
   "id": "b2ea3bfc1415b9b3",
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "开始向Kafka主题 user_profile_tags 发送数据...\n",
      "已生成 1000 条记录\n",
      "已生成 2000 条记录\n",
      "已生成 3000 条记录\n",
      "已生成 4000 条记录\n",
      "已生成 5000 条记录\n",
      "已生成 6000 条记录\n",
      "已生成 7000 条记录\n",
      "已生成 8000 条记录\n",
      "已生成 9000 条记录\n",
      "已生成 10000 条记录\n",
      "已生成 11000 条记录\n",
      "已生成 12000 条记录\n",
      "已生成 13000 条记录\n",
      "已生成 14000 条记录\n",
      "已生成 15000 条记录\n",
      "已生成 16000 条记录\n",
      "已生成 17000 条记录\n",
      "已生成 18000 条记录\n",
      "已生成 19000 条记录\n",
      "已生成 20000 条记录\n",
      "已生成 21000 条记录\n",
      "已生成 22000 条记录\n",
      "已生成 23000 条记录\n",
      "已生成 24000 条记录\n",
      "已生成 25000 条记录\n",
      "已生成 26000 条记录\n",
      "已生成 27000 条记录\n",
      "已生成 28000 条记录\n",
      "已生成 29000 条记录\n",
      "已生成 30000 条记录\n",
      "已生成 31000 条记录\n",
      "已生成 32000 条记录\n",
      "已生成 33000 条记录\n",
      "已生成 34000 条记录\n",
      "已生成 35000 条记录\n",
      "已生成 36000 条记录\n",
      "已生成 37000 条记录\n",
      "已生成 38000 条记录\n",
      "已生成 39000 条记录\n",
      "已生成 40000 条记录\n",
      "已生成 41000 条记录\n",
      "已生成 42000 条记录\n",
      "已生成 43000 条记录\n",
      "已生成 44000 条记录\n",
      "已生成 45000 条记录\n",
      "已生成 46000 条记录\n",
      "已生成 47000 条记录\n",
      "已生成 48000 条记录\n",
      "已生成 49000 条记录\n",
      "已生成 50000 条记录\n",
      "总共发送 50000 条记录\n",
      "数据发送完成！\n",
      "生产者已关闭\n"
     ]
    }
   ],
   "execution_count": 12
  },
  {
   "metadata": {},
   "cell_type": "code",
   "source": "",
   "id": "b10b41ba632e0767",
   "outputs": [],
   "execution_count": null
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
